package com.lcn.netty.mq;

import com.lcn.netty.manager.UserChannelSession;
import com.lcn.pojo.netty.DataContent;
import com.lcn.utils.JsonUtils;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class RabbitMQConnectUtils {

    private final List<Connection> connections = new ArrayList<>();
    private final int maxConnection = 20;

    // 开发环境 dev
    private final String host = "127.0.0.1";
    private final int port = 5672;
    private final String username = "lin";
    private final String password = "lin123";
    private final String virtualHost = "wechat-dev";

    // 生产环境 prod
    //private final String host = "";
    //private final int port = 5672;
    //private final String username = "123";
    //private final String password = "123";
    //private final String virtualHost = "123";

    public ConnectionFactory factory;

    public ConnectionFactory getRabbitMqConnection() {
        return getFactory();
    }

    public ConnectionFactory getFactory() {
        initFactory();
        return factory;
    }

    private void initFactory() {
        try {
            if (factory == null) {
                factory = new ConnectionFactory();
                factory.setHost(host);
                factory.setPort(port);
                factory.setUsername(username);
                factory.setPassword(password);
                factory.setVirtualHost(virtualHost);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void sendMsg(String message, String queue) throws Exception {
        Connection connection = getConnection();
        Channel channel = connection.createChannel();
        channel.basicPublish("",
                            queue,
                            MessageProperties.PERSISTENT_TEXT_PLAIN,
                            message.getBytes("utf-8"));
        channel.close();
        setConnection(connection);
    }

    public void sendMsg(String message, String exchange, String routingKey) throws Exception {
        Connection connection = getConnection();
        Channel channel = connection.createChannel();
        channel.basicPublish(exchange,
                            routingKey,
                            MessageProperties.PERSISTENT_TEXT_PLAIN,
                            message.getBytes("utf-8"));
        channel.close();
        setConnection(connection);
    }

    public GetResponse basicGet(String queue, boolean autoAck) throws Exception {
        GetResponse getResponse = null;
        Connection connection = getConnection();
        Channel channel = connection.createChannel();
        getResponse = channel.basicGet(queue, autoAck);
        channel.close();
        setConnection(connection);
        return getResponse;
    }

    public Connection getConnection() throws Exception {
        return getAndSetConnection(true, null);
    }

    public void setConnection(Connection connection) throws Exception {
        getAndSetConnection(false, connection);
    }

    /**
     * 监听已订阅的消息(广播模式)
     *
     * @param exchangeName 交换机名称
     * @param queueName    队列名称
     */
    public void listen(String exchangeName, String queueName) throws Exception {
        Connection connection = getConnection();
        Channel channel = connection.createChannel();
        /*
         * exchangeName: 交换机名称
         * BuiltinExchangeType.FANOUT: 交换机类型
         * true: 是否持久化
         * false: 是否自动删除
         * null: 参数
         */
        channel.exchangeDeclare(exchangeName,
                BuiltinExchangeType.FANOUT,
                true,
                false,
                null);
        /*
         * queueName: 队列名称
         * true: 是否持久化
         * false: 是否排他性
         * false: 是否自动删除
         * null: 参数
         */
        channel.queueDeclare(queueName,
                true,
                false,
                false,
                null);

        // 队列绑定交换机
        channel.queueBind(queueName, exchangeName, "");
        Consumer consumer = new DefaultConsumer(channel) {
            /**
             * 重写消息交付方法
             * @param consumerTag 消息的标识
             * @param envelope    信封，包括交换机路由信息
             * @param properties  配置信息和内容
             * @param body        收到的消息数据内容
             * @throws IOException
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String exchange = envelope.getExchange();
                String msg = new String(body);
                System.out.println("收到消息exchange = " + exchange+", exchangeName = " + exchangeName);
                if (exchange.equalsIgnoreCase(exchangeName)) {
                    System.out.println("======>消息msg=" + msg);
                    DataContent dataContent = JsonUtils.jsonToPojo(msg, DataContent.class);
                    System.out.println("dataContent = " + dataContent);
                    String senderId = dataContent.getChatMsg().getSenderId(); // 发送者
                    String receiverId = dataContent.getChatMsg().getReceiverId(); // 接收者
                    // 发送给对方
                    List<io.netty.channel.Channel> senderChannels = UserChannelSession.getMultiSession(receiverId);
                    UserChannelSession.sendToTarget(senderChannels, dataContent);

                    // 发送给自己(多端)
                    String myChannelId = dataContent.getExtend();
                    List<io.netty.channel.Channel> myOtherChannels = UserChannelSession.getMyOtherChannel(senderId, myChannelId);
                    UserChannelSession.sendToTarget(myOtherChannels, dataContent);
                }
            }
        };

        /*
         * queueName: 监听的队列名称
         * true: 是否自动确认
         * consumer: 回调函数，处理监听到的消息
         */
        channel.basicConsume(queueName, true, consumer);

    }

    private synchronized Connection getAndSetConnection(boolean isGet, Connection connection) throws Exception {
        getRabbitMqConnection();

        if (isGet) {
            if (connections.isEmpty()) {
                return factory.newConnection();
            }
            Connection newConnection = connections.get(0);
            connections.remove(0);
            if (newConnection.isOpen()) {
                return newConnection;
            } else {
                return factory.newConnection();
            }
        } else {
            if (connections.size() < maxConnection) {
                connections.add(connection);
            }
            return null;
        }
    }

}
